package com.lzq.mq.nameserver.replication;

import com.lzq.mq.common.coder.TcpMsgDecoder;
import com.lzq.mq.common.coder.TcpMsgEncoder;
import com.lzq.mq.common.utils.AssertUtils;
import com.lzq.mq.nameserver.common.CommonCache;
import com.lzq.mq.nameserver.common.MasterSlaveReplicationProperties;
import com.lzq.mq.nameserver.common.NameserverProperties;
import com.lzq.mq.nameserver.enums.ReplicationModeEnum;
import com.lzq.mq.nameserver.enums.ReplicationRoleEnum;
import com.lzq.mq.common.event.EventBus;
import com.lzq.mq.nameserver.event.handler.MasterReplicationServerHandler;
import com.lzq.mq.nameserver.event.handler.SlaveReplicationServerHandler;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

@Slf4j
public class ReplicationService {

    // 参数校验
    public ReplicationModeEnum checkProperties() {

        NameserverProperties nameserverProperties = CommonCache.getNameserverProperties();
        String mode = nameserverProperties.getReplicationMode();
        if (mode.equals(ReplicationModeEnum.STANDALONE.getCode())) {
            log.info("以单机模式启动");
            return null;
        }

        ReplicationModeEnum replicationMode = ReplicationModeEnum.of(mode);
        AssertUtils.isNotNull(replicationMode, "复制模式参数异常");

        if (replicationMode == ReplicationModeEnum.TRACE) {
            // 链路复制
        } else {
            // 主从复制
            MasterSlaveReplicationProperties masterSlaveReplicationProperties = nameserverProperties.getMasterSlaveReplicationProperties();
            AssertUtils.isNotBlank(masterSlaveReplicationProperties.getMaster(), "master参数不能为空");
            AssertUtils.isNotBlank(masterSlaveReplicationProperties.getRole(), "role参数不能为空");
            AssertUtils.isNotBlank(masterSlaveReplicationProperties.getType(), "type参数不能为空");
            AssertUtils.isNotNull(masterSlaveReplicationProperties.getPort(), "同步端口不能为空");
        }

        return replicationMode;
    }

    // 根据参数判断复制的方式
    public void startReplicationTask(ReplicationModeEnum replicationMode) {

        // 单机版本, 无需其他处理
        if (replicationMode == null) {
            return;
        }

        int port = 0;

        if (replicationMode == ReplicationModeEnum.MASTER_SLAVE) {
            // 主从模式, 获取主节点的端口
            port = CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties().getPort();
        }

        ReplicationRoleEnum role = ReplicationRoleEnum.of(CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties().getRole());

        int replicationPort = port;
        Thread  replicationTask = new Thread(() -> {
            if (role == ReplicationRoleEnum.MASTER) {
                EventBus eventBus = new EventBus("主从复制-主节点复制");
                MasterReplicationServerHandler masterReplicationServerHandler = new MasterReplicationServerHandler(eventBus);
                // 启动主节点监听
                startNettyServer(masterReplicationServerHandler, replicationPort);

            } else if (role == ReplicationRoleEnum.SLAVE) {
                // 开启一个
                EventBus eventBus = new EventBus("主从复制-从节点复制");
                SlaveReplicationServerHandler slaveReplicationServerHandler = new SlaveReplicationServerHandler(eventBus);
                String masterAddress = CommonCache.getNameserverProperties().getMasterSlaveReplicationProperties().getMaster();
                startMasterConn(slaveReplicationServerHandler, masterAddress);
            }
        });

        replicationTask.setName("nameserver数据同步进程");
        replicationTask.start();
    }

    // 如果是集群模式, 启动netty进程


    /**
     * 开启对目标进程的连接
     */
    private void startMasterConn(SimpleChannelInboundHandler simpleChannelInboundHandler, String address) {
        // 向主节点发起连接
        EventLoopGroup clientGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        Channel channel;
        bootstrap.group(clientGroup);
        bootstrap.channel(NioSocketChannel.class);
        bootstrap.handler(new ChannelInitializer<SocketChannel>() {
            @Override
            protected void initChannel(SocketChannel ch) throws Exception {
                ch.pipeline().addLast(new TcpMsgDecoder());
                ch.pipeline().addLast(new TcpMsgEncoder());
                ch.pipeline().addLast(simpleChannelInboundHandler);
            }
        });
        ChannelFuture channelFuture = null;
        try {
            Runtime.getRuntime().addShutdownHook(new Thread(() -> {
                clientGroup.shutdownGracefully();
                log.info("nameserver's replication connect application is closed");
            }));
            String[] addr = address.split(":");
            channelFuture = bootstrap.connect(addr[0], Integer.parseInt(addr[1])).sync();
            //连接了master节点的channel对象，建议保存
            channel = channelFuture.channel();
            log.info("成功同上游同步节点建立连接!");
            CommonCache.setMasterConnection(channel);
            channel.closeFuture().sync();
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    /**
     * 启动netty连接
     * @param simpleChannelInboundHandler
     * @param port
     */
    private void startNettyServer(SimpleChannelInboundHandler simpleChannelInboundHandler, int port) {

        //负责netty启动
        NioEventLoopGroup bossGroup = new NioEventLoopGroup();
        //处理网络io中的read&write事件
        NioEventLoopGroup workerGroup = new NioEventLoopGroup();
        ServerBootstrap bootstrap = new ServerBootstrap();
        bootstrap.group(bossGroup, workerGroup);
        bootstrap.channel(NioServerSocketChannel.class);
        bootstrap.childHandler(new ChannelInitializer<Channel>() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ch.pipeline().addLast(new TcpMsgDecoder());
                ch.pipeline().addLast(new TcpMsgEncoder());
                ch.pipeline().addLast(simpleChannelInboundHandler);
            }
        });
        Runtime.getRuntime().addShutdownHook(new Thread(() -> {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
            log.info("nameserver's replication application is closed");
        }));
        ChannelFuture channelFuture = null;
        try {
            channelFuture = bootstrap.bind(port).sync();
            log.info("start nameserver's replication application on port:" + port);
            //阻塞代码
            channelFuture.channel().closeFuture().sync();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }
}
